package com.study.chapter11;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @Description:
 * @Author: LiuQun
 * @Date: 2022/8/19 21:31
 */
public class CreateTableTest {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //创建表环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);


        //创建实体表
        String createDDL = " CREATE TABLE click_table (" +
                " `user` STRING, " +
                " url STRING, " +
                " ts BIGINT " +
                " ) WITH ( " +
                " 'connector' = 'filesystem', " +   //指定连接器为文件
                " 'path' = 'input/cart.txt', " +    //指定文件路径
                " 'format' = 'csv' " +              //指定格式
                " ) ";

        tableEnv.executeSql(createDDL);

        //使用Table API进行表的查询转换，从表环境中获取Table
        Table clickTable = tableEnv.from("click_table");
        Table resultTab = clickTable.where($("user").isEqual("Bob"))
                .select($("user"), $("url"));

        //注册一张虚拟表
        tableEnv.createTemporaryView("clickTable",clickTable);
        tableEnv.createTemporaryView("result",resultTab);

        //执行sql进行表的查询转换：下面result要加引号，否则会报错，因为这个result是sql定义的关键字
        Table result = tableEnv.sqlQuery("select url,user from `result` ");

        //创建一张用于输出的表
        String createOutDDL = " CREATE TABLE click_out_table (" +
                " `user` STRING, " +
                " url STRING " +
                " ) WITH ( " +
                " 'connector' = 'filesystem', " +   //指定连接器为文件
                " 'path' = 'output', " +            //指定输出的文件路径
                " 'format' = 'csv' " +              //指定格式
                " ) ";
        tableEnv.executeSql(createOutDDL);
        //将结果输出到表
        result.executeInsert("click_out_table");

        Table userNumTab = tableEnv.sqlQuery("select user,count(url) as num from clickTable group by user");

        //创建一张用于控制台打印输出的表
        String createPrintDDL = " CREATE TABLE print_table (" +
                " `user` STRING, " +
                " num BIGINT " +
                " ) WITH ( " +
                " 'connector' = 'print' " +   //指定连接器为控制台打印
                " ) ";
        tableEnv.executeSql(createPrintDDL);
        userNumTab.executeInsert("print_table");
    }
}
